package com.shujia.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}

object Demo2WordCount2 {
  def main(args: Array[String]): Unit = {
    /**
     * Spark core: SparkContext 核心数据结构：RDD
     * Spark sql: SparkSession 核心数据结构：DataFrame
     * Spark streaming: StreamingContext  核心数据结构：DStream(底层封装了RDD)
     */

    val conf = new SparkConf()
    conf.setMaster("local[2]") // 给定核数
    conf.setAppName("spark Streaming 单词统计")
    val sparkContext = new SparkContext(conf)

    //创建Spark Streaming的运行环境，和前两个模块是不一样的
    //Spark Streaming是依赖于Spark core的环境的
    //this(sparkContext: SparkContext, batchDuration: Duration)
    //Spark Streaming处理之前，是有一个接收数据的过程
    //batchDuration，表示接收多少时间段内的数据
    val streamingContext = new StreamingContext(sparkContext, Durations.seconds(5))
    //设置的是一个文件夹
    streamingContext.checkpoint("spark/data/checkpoint2")

    //Spark Streaming程序理论上是一旦启动，就不会停止，除非报错，人为停止，停电等其他突然场景导致程序终止
    //监控一个端口号中的数据，手动向端口号中打数据
    val rids: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)
    //hello world

    val wordsDS: DStream[String] = rids.flatMap(_.split(" "))
    val kvDS: DStream[(String, Int)] = wordsDS.map((_, 1)) // (hello,1) (hello,1)  (hello,1)

    /**
     * 每5秒中resDS中的数据，是当前5s内的数据
     * reduceByKey，只会对当前5s批次中的数据求和
     */
    //    val resDS: DStream[(String, Int)] = kvDS.reduceByKey(_ + _)

    /**
     * 需要使用有状态的算子来处理当前批次数据与历史数据的关系
     */
    /**
     * updateStateByKey[S: ClassTag](updateFunc: (Seq[V], Option[S]) => Option[S]): DStream[(K, S)]
     * Seq:  序列，表示历史键对应的值组成的序列 (hello, seq:[1,1,1])
     * Option: 当前批次输入键对应的value值，如果历史中没有该键，这个值就是None, 如果历史中出现了这个键，这个值就是Some(值)
     *
     * 有状态算子使用注意事项：
     * 1、有状态算子ByKey算子只适用于k-v类型的DStream
     * 2、有状态算子使用的时候，需要提前设置checkpoint的路径，因为需要将历史批次的结果存储下来
     */
    val resDS: DStream[(String, Int)] = kvDS.updateStateByKey((seq1: Seq[Int], opt1: Option[Int]) => {
      val sumValue: Int = seq1.sum
      val num: Int = opt1.getOrElse(0)
      Option(sumValue + num)
    })


    println("--------------------------------------")
    resDS.print()
    println("--------------------------------------")

    /**
     * sparkStreaming启动的方式和前两个模块启动方式不一样
     */
    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()

  }
}
